package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.function.AsyncDimFunction;
import com.atguigu.gmall.realtime.function.DorisMapFunction;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.DorisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/8/11 08:50
 */
public class Dws_09_DwsTradeSkuOrderWindow_Async extends BaseApp {
    public static void main(String[] args) {
        new Dws_09_DwsTradeSkuOrderWindow_Async().start(
            40009,
            2,
            "Dws_09_DwsTradeSkuOrderWindow",
            GmallConstant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 封装数据到 pojo 中
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream = parseToPojo(stream);
        
        // 2. 按照详情 id 实现去重
        beanStream = distinctByOrderDetailId(beanStream);
        
        // 3. 开窗聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithoutDims = windowAndAgg(beanStream);
        
        // 4. 补充维度信息
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithDims = addDims(beanStreamWithoutDims);
        
        
        // 5. 写出到 doris 中
        writeToDoris(beanStreamWithDims);
        
        
    }
    
    private void writeToDoris(SingleOutputStreamOperator<TradeSkuOrderBean> beanStreamWithDims) {
        beanStreamWithDims
            .map(new DorisMapFunction<>())
            .sinkTo(DorisUtil.getDorisSink("gmall2023.dws_trade_sku_order_window"));
        
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> addDims(
        SingleOutputStreamOperator<TradeSkuOrderBean> stream) {
        
        SingleOutputStreamOperator<TradeSkuOrderBean> skuStream = AsyncDataStream
            .unorderedWait(stream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_sku_info";
                               }
                
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getSkuId();
                               }
                
                               @Override
                               public void addDim(TradeSkuOrderBean bean, JSONObject dim) {
                    
                                   bean.setSkuName(dim.getString("sku_name"));
                                   bean.setSpuId(dim.getString("spu_id"));
                                   bean.setTrademarkId(dim.getString("tm_id"));
                                   bean.setCategory3Id(dim.getString("category3_id"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> spuStream = AsyncDataStream
            .unorderedWait(skuStream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_spu_info";
                               }
                
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getSpuId();
                               }
                
                               @Override
                               public void addDim(TradeSkuOrderBean bean,
                                                  JSONObject dim) {
                                   bean.setSpuName(dim.getString("spu_name"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        
        SingleOutputStreamOperator<TradeSkuOrderBean> tmStream = AsyncDataStream
            .unorderedWait(spuStream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_base_trademark";
                               }
                
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getTrademarkId();
                               }
                
                               @Override
                               public void addDim(TradeSkuOrderBean bean,
                                                  JSONObject dim) {
                                   bean.setTrademarkName(dim.getString("tm_name"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream
            .unorderedWait(tmStream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_base_category3";
                               }
                
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getCategory3Id();
                               }
                
                               @Override
                               public void addDim(TradeSkuOrderBean bean,
                                                  JSONObject dim) {
                                   bean.setCategory3Name(dim.getString("name"));
                                   bean.setCategory2Id(dim.getString("category2_id"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream
            .unorderedWait(c3Stream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_base_category2";
                               }
                
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getCategory2Id();
                               }
                
                               @Override
                               public void addDim(TradeSkuOrderBean bean,
                                                  JSONObject dim) {
                                   bean.setCategory2Name(dim.getString("name"));
                                   bean.setCategory1Id(dim.getString("category1_id"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        return AsyncDataStream
            .unorderedWait(c2Stream,
                           new AsyncDimFunction<TradeSkuOrderBean>() {
                               @Override
                               public String getTableName() {
                                   return "dim_base_category1";
                               }
    
                               @Override
                               public String getId(TradeSkuOrderBean bean) {
                                   return bean.getCategory1Id();
                               }
    
                               @Override
                               public void addDim(TradeSkuOrderBean bean,
                                                  JSONObject dim) {
                                   bean.setCategory1Name(dim.getString("name"));
                               }
                           },
                           120,
                           TimeUnit.SECONDS
            );
        
        
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> windowAndAgg(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeSkuOrderBean::getSkuId)
            .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1,
                                                    TradeSkuOrderBean value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                        value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                        
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String skuId,
                                        Context ctx,
                                        Iterable<TradeSkuOrderBean> elements,
                                        Collector<TradeSkuOrderBean> out) throws Exception {
                        TradeSkuOrderBean bean = elements.iterator().next();
                        bean.setStt(DateFormatUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(DateFormatUtil.tsToDateTime(ctx.window().getEnd()));
                        bean.setCurDate(DateFormatUtil.tsToDate(ctx.window().getStart()));
                        
                        out.collect(bean);
                    }
                }
            );
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> distinctByOrderDetailId(
        SingleOutputStreamOperator<TradeSkuOrderBean> beanStream) {
        /*
        为什么会有重复数据? 数据是由于 left join 写入的结果
            详情id    金额      活动表相关的金额     优惠券相关的金额          生成的时间
             1       100        null                null                    1
             null(删除)
             1       100        200                 null                    2
             null(删除)
             1       100        200                 300                     3
             
          用流消费:
            详情id    金额      活动表相关的金额     优惠券相关的金额
             1       100        null                null
             1       100        200                 null
             1       100        200                 300
            导致同一个详情出现多次, 后面进行聚和运算的时候会出现重复计算, 所以需要去重
            
            
            同一个详情,只保留最后一个最完整的数据
            
            思路1: 会话窗口
                假设 dwd 的数据添加一个数据的生成时间,则时间最大的那个一定是数据最全,我们想要的.
                   详情id    金额      活动表相关的金额     优惠券相关的金额   生成的时间
                     1       100        null                null             1
                     1       100        200                 null             2
                     1       100        200                 300              3
                     
                 按照 sku 分区,设置会话窗口 窗口 gap 5s
                    当同一个详情的最后一条数据到了之后, 等待 5s 后, 窗口关闭,按照时间进行排序,取出时间最大的就是数据最完整
                    
                    优点:
                        简单,实现方便
                    缺点:
                        时效性低: 最后一条数据到,5s之后才会出结果
                        
             思路 2: 定时器
                当第一条数据来了之后,注册一个 5s 后触发定时器, 当定时器触发的时候,则数据一定全部来齐了
                    每来一条就比较一下时间大小,把时间大的存入到状态中,当定时器触发的时候,转态中存储就一定是时间最大的那个
                    
                 优点:
                     简单,实现方便
                 缺点: 时效比较会话窗口高,但是还是有实效性问题: 第一条到了之后等到 5s 出结果
                 
              思路3: 抵消法
                
                             详情id    金额      活动表相关的金额     优惠券相关的金额
                   第一条      1       100        0                 0             直接输出,  存如到状态
                              
                              1       -100       0                 0              给第一条取反输出
                   第二条      1       100        200                0             直接输出
                   
                              1       -100       -200                0           给第二条取反输出
                   第三条      1       100        200                 300          直接输出
           
                    优点: 实效性最高
                    缺点: 写放大效应
                    
                 优化:
                            详情id    金额         活动表相关的金额        优惠券相关的金额
                   第一条      1       100           0                           0             直接输出,  存如到状态
              
                   第二条      1       100-100        200-0                     0-0            直接输出  存到状态(第二条原始数据)
                   
                   第三条      1       100-100         200-200                 300-0            直接输出  存到状态(第三条原始数据)
                   
                   优点: 实效性最高, 没有写放大
                 思路4: 特殊情况
                       当如果需要的数据都在左表,则第一条数据就有了所有的数据了.直接输出即可.
                       
                        找到第一条: 状态为 null 的时候
            
         */
        return beanStream
            .keyBy(TradeSkuOrderBean::getOrderDetailId)
            .process(new KeyedProcessFunction<String, TradeSkuOrderBean, TradeSkuOrderBean>() {
                
                private ValueState<TradeSkuOrderBean> lastBeanState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<TradeSkuOrderBean> desc = new ValueStateDescriptor<>("lastBean", TradeSkuOrderBean.class);
                    StateTtlConfig ttlConf = StateTtlConfig.newBuilder(Time.seconds(50000))
                        .updateTtlOnCreateAndWrite()
                        .neverReturnExpired()
                        .build();
                    desc.enableTimeToLive(ttlConf);
                    lastBeanState = getRuntimeContext().getState(desc);
                }
                
                @Override
                public void processElement(TradeSkuOrderBean bean,
                                           Context ctx,
                                           Collector<TradeSkuOrderBean> out) throws Exception {
                    TradeSkuOrderBean lastBean = lastBeanState.value();
                    
                    if (lastBean == null) { // 第一条
                        // 当前数据直接输出
                        out.collect(bean);
                    } else {  // 不是第一条
                        // 上一条数据取反 + 当前数据, 然后输出 或者 当前 - 上一条
                        lastBean.setOrderAmount(bean.getOrderAmount().subtract(lastBean.getOrderAmount()));
                        lastBean.setOriginalAmount(bean.getOriginalAmount().subtract(lastBean.getOriginalAmount()));
                        lastBean.setActivityAmount(bean.getActivityAmount().subtract(lastBean.getActivityAmount()));
                        lastBean.setCouponAmount(bean.getCouponAmount().subtract(lastBean.getCouponAmount()));
                        out.collect(lastBean);
                    }
                    
                    // 把当前数据存入到状态
                    lastBeanState.update(bean);
                    
                    
                }
            });
    }
    
    private SingleOutputStreamOperator<TradeSkuOrderBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
            .map(new MapFunction<String, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(String jsonStr) throws Exception {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    
                    return TradeSkuOrderBean.builder()
                        .orderDetailId(obj.getString("id"))
                        .skuId(obj.getString("sku_id"))
                        .orderAmount(obj.getBigDecimal("split_total_amount"))
                        .activityAmount(obj.getBigDecimal("split_activity_amount"))
                        .couponAmount(obj.getBigDecimal("split_coupon_amount"))
                        .originalAmount(obj.getBigDecimal("split_original_amount"))
                        .ts(obj.getLong("ts") * 1000)
                        .build();
                }
                
            });
    }
}
/*
异步超时异常:
    1. 检查所有集群是否正常
            hdfs hbase redis kafka
    2. 再检测,用到的 6 张表是否都正常, 数据是否完整
    
    3. 加大超时时间

-------
 默认情况, 所有的算子同步的方式处理数据
    同步: 对单一并行度来说, 如果一个元素没有处理完,则不会处理下一个元素
    
        不能充分的利用网络等待时间
        
 如何解决:
    1. 增加并行度
        成本比较高 占据独立的内存, 是有上限的
        
    2. 单个并行度开启异步的方式处理元素
    
-----
1. flink的异步 io + 异步客户端(一个并行度一个) + java 8 的异步框架

2. 如果没有异步客户端怎么办?
    flink的异步 io + 多线程(线程池) + 多同步客户端
        一个线程一个客户端
*/